package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo02UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    /**
     * 有状态算子：UpdateStateByKey
     * 可以做到将当前结果和之前的状态整合在一起形成新的状态
     */

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]")
      .getOrCreate()

    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))
    ssc.checkpoint("spark/data/ck")

    val textDS: DStream[String] = ssc.socketTextStream("master", 8888)

    textDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      /*
      会对Key进行分组统计
      分组之后，会将同一个Key的当前批次里对应的所有Value构建成Seq
      opt实际上对应上一个批次中当前Key计算出来的结果，
      如果是第一个批次，则没有上一个批次，所以类型为Option
       */
      // 如果使用了有状态算子，则需要指定Checkpoint的目录
      .updateStateByKey((seq: Seq[Int], opt: Option[Int]) => {
        // Seq:当前批次 某一个Key 对应的 所有Value
        val seqSum: Int = seq.sum // 当前批次的结果
        opt match {
          case Some(v) =>
            Some(v + seqSum) // 加上上一次的结果
          case None =>
            Some(seqSum)
        }
      })
      .print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

}
